package parallels.servertech.server.nio.event;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.logging.Logger;

import parallels.servertech.server.nio.event.ChannelEvent.ChannelEventType;

public class CommandDecoder implements ChannelEventHandler {
	private static final Logger log = Logger.getLogger("command-processor");
	
	private enum CommandProcessingState {
		READY_TO_READ, READING_CMD_NAME_LEN, READING_CMD_NAME, READING_CMD_ARGS_LEN, READING_CMD_ARGS, WRITING_RES
	}
	
	// TODO add sum command support and create non-interactive client for sum command testing
	private CommandProcessingState state;
	
	private final ByteArrayOutputStream utfBuf = new ByteArrayOutputStream();
	private final ChannelContext context = new  ChannelContext();
	
	private ByteBuffer cmdSectionBuf;
	private ByteBuffer resultBuf;
	
	private String cmdName;
	
	public CommandDecoder() {
		state = CommandProcessingState.READY_TO_READ;
	}

	@Override
	public void channelReadable(ChannelEvent event) throws IOException {
		ByteBuffer channelBuf = ByteBuffer.allocate(4096);
		utfBuf.reset();
		SocketChannel channel = event.getChannel();
		context.setBuffer(channelBuf);
		ChannelEventDispatcher dispatcher = event.getEventDispatcher();
		dispatcher.sendReply(new ChannelEvent(dispatcher, ChannelEventType.INPUT_BUFFER_READY_FOR_FILLING, channel, event.getSelectionKey()));
	}

	@Override
	public void inputBufferFilled(ChannelEvent event) throws IOException {
		ByteBuffer inputBuffer = context.getBuffer();
		log.info(inputBuffer.remaining() + " bytes available for processing");
		
		switch (state) {
		case READY_TO_READ:
			changeState(CommandProcessingState.READING_CMD_NAME_LEN, Short.BYTES);
			readCommandNameBytesLenght(event);
			break;
		case READING_CMD_NAME_LEN:
			readCommandNameBytesLenght(event);
			break;
		case READING_CMD_NAME:
			readCommandName(event);
			break;
		case READING_CMD_ARGS:
			readCommandArguments(event);
			break;
		default:
			throw new IllegalStateException("unexpected state: " + state);	
		}
	}
	
	@Override
	public void outputBufferDrained(ChannelEvent event) throws IOException {
		changeState(CommandProcessingState.READY_TO_READ, 0);
	}
	
	@Override
	public ChannelContext getChannelContext() {
		return context;
	}

	private void readCommandNameBytesLenght(ChannelEvent event) throws IOException {
		log.info("reading command name length");
		boolean readFully = readCommandSectionBytes(context.getBuffer());
		if (readFully) {
			cmdSectionBuf.flip();
			short cmdNameLen = cmdSectionBuf.getShort();
			log.info("command name length = " + cmdNameLen);
			changeState(CommandProcessingState.READING_CMD_NAME, cmdNameLen);
			readCommandName(event);
		}
	}
	
	private void readCommandName(ChannelEvent event) throws IOException {
		log.info("reading command name");
		boolean readFully = readCommandSectionBytes(context.getBuffer());
		if (readFully) {
			try (ByteArrayInputStream cmdNameUtf8 = new ByteArrayInputStream(utfBuf.toByteArray(), 0, Short.BYTES + cmdSectionBuf.limit());
					DataInputStream cmdNameInput = new DataInputStream(cmdNameUtf8)) {
				
				cmdName = cmdNameInput.readUTF();
				log.info("command name = " + cmdName);
			}
			
			switch (cmdName) {
			case "add":
				// TODO consider extracting this logic into command handlers
				changeState(CommandProcessingState.READING_CMD_ARGS, 2 * Integer.BYTES);
				readCommandArguments(event);
				break;
			case "shutdown":
				event.getEventDispatcher().stop();
				break;
			default:
				throw new IllegalStateException("unsupported command: " + cmdName);
			}
		}
	}
	
	private void readCommandArguments(ChannelEvent event) {
		log.info("reading " + cmdName + " command arguments");
		
		boolean readFully = readCommandSectionBytes(context.getBuffer());
		if (readFully) {
			switch (cmdName) {
			case "add":
				// calculating result
				cmdSectionBuf.flip();
				int left = cmdSectionBuf.getInt();
				int right = cmdSectionBuf.getInt();
				
				resultBuf = ByteBuffer.allocate(2 * Integer.BYTES);
				// status code
				resultBuf.putInt(0);
				resultBuf.putInt(left + right);
				
				ChannelEventDispatcher dispatcher = event.getEventDispatcher();
				resultBuf.flip();
				context.setBuffer(resultBuf);
				dispatcher.sendReply(new ChannelEvent(dispatcher, ChannelEventType.OUTPUT_BUFFER_READY_FOR_DRAINING, event.getChannel(), event.getSelectionKey()));
				break;
			default:
				throw new IllegalStateException("unsupported command: " + cmdName);	
			}
		}
	}
	
	private boolean readCommandSectionBytes(ByteBuffer buffer) {
		if (buffer.remaining() >= cmdSectionBuf.remaining()) {
			// saving source buffer limit
			int bufLimit = buffer.limit();
			// reading remaining section bytes
			buffer.limit(buffer.limit() - (buffer.remaining() - cmdSectionBuf.remaining()));
			cmdSectionBuf.put(buffer);
			// restoring source buffer limit to allow reading remaining bytes from it
			buffer.limit(bufLimit);			
			// TODO UTF buffer should be filled only when necessary
			utfBuf.write(cmdSectionBuf.array(), 0, cmdSectionBuf.limit());
			return true;
		} else {
			cmdSectionBuf.put(buffer);
			return false;
		}
	}

	private void changeState(CommandProcessingState nextState, int sectionBufLen) {
		if (sectionBufLen > 0) {
			cmdSectionBuf = ByteBuffer.allocate(sectionBufLen);
		}
		state = nextState;
	}
}
